package com.xinqing.bigdata.test.datastream

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011


object Kafka2Mysql {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(4)


    // 创建kafka相关的配置 source
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "10.201.7.175:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val inputStream = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

    // transform 算子
    val sourceDataStream = inputStream.map(data => {
      val dataArr = data.split(",")
      SensorReading(dataArr(0), dataArr(1).trim.toLong, dataArr(2).trim.toDouble)
    })

    sourceDataStream.print("data stream")

    sourceDataStream.addSink(new MyJdbcSinkFun2())

    env.execute("jdbc sink test")
  }


}

class MyJdbcSinkFun2 extends RichSinkFunction[SensorReading] {

  var conn: Connection = _
  var insertStatment: PreparedStatement = _
  var updateStatment: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://10.201.7.174:3307/test", "PT_super_user", "Hb86be03cc_H3,,2c8ff86#326c3")
    insertStatment = conn.prepareStatement("insert into temperatures(id,temperature) values(?,?)")
    updateStatment = conn.prepareStatement("update temperatures set temperature=? where id=?")
  }

  override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
    updateStatment.setDouble(1, value.temperature)
    updateStatment.setString(2, value.id)
    val updateCount = updateStatment.executeUpdate()
    if (updateCount == 0) {
      insertStatment.setString(1, value.id)
      insertStatment.setDouble(2, value.temperature)
      insertStatment.execute()
    }
  }

  override def close(): Unit = {
    insertStatment.close()
    updateStatment.close()
    conn.close()
  }
}
case class SensorReading(id: String, timestamp: Long, temperature: Double)

